package org.huang.rocketmq.demo.transaction;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //模拟本地事务，所有的事务都是UNKNOW。这意味着所有的事务消息都要进行回查
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        System.out.println("本地事务TransactionId="+msg.getTransactionId()+",status="+status);
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        //根据事务id查询当前的事务状态是否完成。
        Integer status = localTrans.get(msg.getTransactionId());
        System.out.println("回查事务TransactionId="+msg.getTransactionId()+",status="+status);
        if (null != status) {
            switch (status) {
                case 0:
                    //事务还没有结束，需要下次再来查询
                    return LocalTransactionState.UNKNOW;
                case 1:
                    //事务已经完成，该消息可以被消费者消费
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    //事务已经失败，该消息将不被消费者消费
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                default:
                    return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
